package profiterole.mapreduce;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;

import profiterole.api.Waffle;
import profiterole.waffle.WaffleImpl;

public class Reducer {

	// seems major bottle neck here specially if the map is huge
	// thus need to look for problems with small maps
	// such as seach/grep or word count
	// can add heuristics for hash table size

	// Reducer(when we pass to it identity func) is not reducer
	// Do limits for reducer, sort of hash must be less than 50,

	private Reducer() {
		// no need for constructor 1 object/function per JVM
	}

	public static Waffle<Integer> reduce(List<HashMap<String, Integer>> maps) {
		HashMap<String, Integer> accumMap  = new HashMap<String, Integer>();

		// library code check parameters for validity
		if (maps == null) {

			// TODO do empty waffle
			return null;
		}

		// TODO current problem works only for ints use generic reduce version
		for (HashMap<String, Integer> map : maps) {

			for (String word : map.keySet()) {
				if (accumMap.containsKey(word)) {
					int value = map.get(word).intValue()
							+ accumMap.get(word).intValue();
					accumMap.put(word, value);
				} else {
					accumMap.put(word, map.get(word).intValue());
				}
			}
		}

		return new WaffleImpl<Integer>(accumMap);
	}

	// FUTURE DIRECTION
	// List-based generic reduction
	static <E> E reduce(List<E> list, Function<E> f, E initVal) {
		List<E> snapshot;
		synchronized (list) {
			snapshot = new ArrayList<E>(list);
		}
		E result = initVal;
		for (E e : snapshot)
			result = f.apply(result, e);
		return result;
	}

	interface Function<M> {
		M apply(M arg1, M arg2);
	}
}